package com.cloud.bssp.java8.forkjoinpool;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
import java.util.stream.LongStream;

import static com.cloud.bssp.java8.stream.TestStreamParallel.measureSumPerf;

/**
 * @description： 使用ForkJoinPool
 * @author：weirx
 * @date：2021/10/25 14:10
 * @version：3.0
 */
public class TestRecursiveTask extends RecursiveTask<Long> {

    /**
     * 要求和的数组
     */
    private final long[] numbers;

    /**
     * 子任务求和的数组的开始位置
     */
    private int start;

    /**
     * 子任务求和的数组的结束位置
     */
    private int end;

    /**
     * 私有构造，用于以递归方式为主任务创建子任务
     *
     * @param numbers
     * @param start
     * @param end
     */
    private TestRecursiveTask(long[] numbers, int start, int end) {
        this.numbers = numbers;
        this.start = start;
        this.end = end;
    }

    /**
     * 公共函数用于构建主任务
     *
     * @param numbers
     */
    public TestRecursiveTask(long[] numbers) {
        this.numbers = numbers;
    }

    /**
     * 任务拆分的数组最大值
     */
    public static final long THRESHOLD = 10000L;

    @Override
    protected Long compute() {
        int length = end - start;
        if (length <= THRESHOLD) {
            // 如果大小小于等于阈值，则顺序计算
            return computeSequentially();
        } else {
            //创建一个子任务，为数组的前一半求和
            TestRecursiveTask left = new TestRecursiveTask(numbers, start, start + length / 2);
            //利用另一个ForkJoinPool线程异步执行新创建的子任务
            left.fork();
            //创建一个子任务，为数组的后一半求和
            TestRecursiveTask right = new TestRecursiveTask(numbers, start + length / 2, end);
            // 同步执行第二个子任务
            Long compute = right.compute();
            //读取第一个子任务的结果，没有完成则等待
            Long join = left.join();
            //结果合并
            return compute + join;
        }
    }

    /**
     * 当子任务不可拆分时计算结果的简单算法
     *
     * @return
     */
    private Long computeSequentially() {
        long sum = 0;
        for (int i = start; i < end; i++) {
            sum += numbers[i];
        }
        return sum;
    }

    /**
     * 并行对前n个自然数求和
     *
     * @param n
     * @return
     */
    public static long forkJoinSum(long n) {
        long[] numbers = LongStream.rangeClosed(1, n).toArray();
        ForkJoinTask<Long> task = new TestRecursiveTask(numbers);
        return new ForkJoinPool().invoke(task);
    }

    public static void main(String[] args) {
        System.out.println("ForkJoin sum done in: " + measureSumPerf(
                TestRecursiveTask::forkJoinSum, 10000000) + " msecs");
    }

}
